package com.jscloud.bigdata.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Properties;

/**
 *  ### 指定分区数据进行消费
 *  因为每个topic都可能有多个分区，所以我们也可以针对指定的分区进行消费
 */
public class ConsumPartition {
        public static void main(String[] args) {
                Properties props= new Properties();
                props.put("bootstrap.servers","localhost:9092"); props.put("group.id", "test");
                props.put("enable.auto.commit","true");
                props.put("auto.commit.interval.ms","1000");
                props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
                props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
                KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
                String topic ="foo";
                TopicPartition partition0 = new TopicPartition(topic, 0);
                TopicPartition partition1 = new TopicPartition(topic, 1);
                consumer.assign(Arrays.asList(partition0, partition1)); //使用assign来订阅一些partition的数据

                //从某个分区里面指定的offset开始消费
                //TopicPartition topicPartition = new TopicPartition("test",0);
               //consumer.seek(topicPartition, 100); //使用seek来消费指定topic，指定的patition的分区的数据

                //使用subscribe，指定用于某些topic来消费
                //consumer.subscribe(Arrays.asList("test"));

                //手动指定消费指定分区的数据---end
                while (true) {
                        ConsumerRecords<String,String> records = consumer.poll(100);
                        for(ConsumerRecord<String, String> record : records) {
                                System.out.printf("offset= %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                        }

                        consumer.commitSync(); // 表示同步的提交offset的值
                        consumer.commitAsync(); //表示异步的提交offset的值
                }
        }
}